package gora.cascading.example1;

import cascading.flow.FlowDef;
import cascading.flow.hadoop.HadoopFlowConnector;
import cascading.pipe.Pipe;
import cascading.scheme.hadoop.TextDelimited;
import cascading.tap.Tap;
import cascading.tap.hadoop.Hfs;

/**
 * A HadoopFlowConnector is used to Copy from A to B.
 * 
 */
public final class CopyStream {
	
 /**
   * Static method to Copy inPath to outPath. This will be executed in Hadoop cluster
   *
   * @param input source of the file
   * @param place where you want it to be copied
   */
	public static void copy(String inPath, String outPath) {
	HadoopFlowConnector flowConnector = new HadoopFlowConnector();

    // create the source tap
    Tap inTap = new Hfs( new TextDelimited( true, "\t" ), inPath );

    // create the sink tap
    Tap outTap = new Hfs( new TextDelimited( true, "\t" ), outPath );

    // specify a pipe to connect the taps
    Pipe copyPipe = new Pipe( "copy" );

    // connect the taps, pipes, etc., into a flow
    FlowDef flowDef = FlowDef.flowDef()
     .addSource( copyPipe, inTap )
     .addTailSink( copyPipe, outTap );

    // run the flow
    flowConnector.connect( flowDef ).complete();
	}
}
